package my;

import my.common.File;
import my.epoll.EventArray;
import my.epoll.Native;
import my.socket.AbstractSocket;
import my.socket.Socket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;

/**
 * @author fengbo
 */
public class EpollEventLoop implements Executor {

    private static final Logger log = LoggerFactory.getLogger(EpollEventLoop.class);

    private final Queue<Runnable> taskQueue;
    private final int efd;
    private final int tfd;
    private final int vfd;
    private final EventArray events;
    private final Executor executor;

    private final Map<Integer, Socket> channels = new ConcurrentHashMap<>();
    private ThreadLocal<ByteBufferAllocator> cache = new ThreadLocal<>();

    private volatile Thread thread;

    public EpollEventLoop(Executor executor) {
        taskQueue = newTaskQueue(16);
        boolean success = false;
        int efd = 0;
        int tfd = 0;
        int vfd = 0;
        try {
            this.efd = efd = Native.epollCreate();
            this.vfd = vfd = Native.eventCreate();
            try {
                Native.epollCtlAdd(efd, vfd, Native.EPOLLIN);
            } catch (IOException e) {
                throw new IllegalStateException("Unable to add eventFd filedescriptor to epoll", e);
            }
            this.tfd = tfd = Native.timerCreate();
            try {
                Native.epollCtlAdd(efd, tfd, Native.EPOLLIN | Native.EPOLLET);
            } catch (IOException e) {
                throw new IllegalStateException("Unable to add timerFd filedescriptor to epoll", e);
            }
            success = true;
        } finally {
            if (!success) {
                if (efd > 0) {
                    try {
                        File.close(efd);
                    } catch (Exception e) {
                        // ignore
                    }
                }
                if (tfd > 0) {
                    try {
                        File.close(tfd);
                    } catch (Exception e) {
                        // ignore
                    }
                }
                if (vfd > 0) {
                    try {
                        File.close(vfd);
                    } catch (Exception e) {
                        // ignore
                    }
                }
            }
        }
        events = new EventArray(4096);
        this.executor = executor;
    }

    private Queue<Runnable> newTaskQueue(int maxPendingTasks) {
        return new LinkedBlockingQueue<>(maxPendingTasks);
    }

    public boolean inEventLoop(Thread thread) {
        return thread == this.thread;
    }

    @Override
    public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        boolean inEventLoop = inEventLoop(Thread.currentThread());
        addTask(task);
        if (!inEventLoop && thread == null) {
            assert thread == null;
            executor.execute(() -> {
                thread = Thread.currentThread();
                EpollEventLoop.this.run();
            });
        }
    }

    private void run() {
        for (;;) {
            try {
                int ready = epollWait();
                try {
                    if (ready > 0) {
                        processReady(events, ready);
                    }
                } finally {
                    runAllTasks();
                }
            } catch (Throwable e) {
                log.warn("Unexpected exception in the selector loop.", e);

                // Prevent possible consecutive immediate failures that lead to
                // excessive CPU consumption.
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    // Ignore.
                }
            }
        }
    }

    private int epollWait() throws IOException {
        return Native.epollWait(efd, events, tfd);
    }

    private void addTask(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        if (!taskQueue.offer(task)) {
            throw new RejectedExecutionException();
        }
    }

    private void processReady(EventArray events, int ready) throws IOException {
        for (int i = 0; i < ready; i++) {
            long ev = events.events(i);
            int fd = events.fd(i);
            if (vfd == fd) {
                Native.eventFdRead(vfd);
                System.out.println("event read");
                break;
            }
            Socket socket = channels.get(fd);
            if ((ev & (Native.EPOLLERR | Native.EPOLLIN)) != 0) {
                Native.epollCtlMod(efd, fd, my.epoll.Native.EPOLLOUT);
                socket.inReady();
            } else if ((ev & (Native.EPOLLERR | Native.EPOLLOUT)) != 0) {
                socket.outReady();
                Native.epollCtlMod(efd, fd, my.epoll.Native.EPOLLIN);
            }
        }
    }

    private void runAllTasks() {
        Runnable task = taskQueue.poll();
        for (;;) {
            if (task == null) {
                return;
            }
            try {
                task.run();
            } catch (Throwable t) {
                log.warn("A task raised an exception. Task: {}", task, t);
            }
            task = taskQueue.poll();
        }
    }

    public void add(Socket socket) throws IOException {
        assert inEventLoop(Thread.currentThread());
        channels.put(socket.intValue(), socket);
        Native.epollCtlAdd(efd, socket.intValue(), Native.EPOLLIN | Native.EPOLLET);
    }

    void shutdown() {
        cache.remove();
        if (tfd > 0) {
            File.close(tfd);
        }
        if (efd > 0) {
            File.close(efd);
        }
    }

    void remove(AbstractSocket socket) throws IOException {
        assert inEventLoop(Thread.currentThread());

        int fd = socket.intValue();
        if (channels.remove(fd) != null) {
            // Remove the epoll. This is only needed if it's still open as otherwise it will be automatically
            // removed once the file-descriptor is closed.
            Native.epollCtlDel(efd, fd);
        }
    }

    public ByteBufferAllocator allocator() {
        if (cache.get() == null) {
            cache.set(new ByteBufferAllocator());
        }
        return cache.get();
    }

    public int getVfd() {
        return vfd;
    }
}
